﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace WindowsService1
{
    public class MqttMessageQueue
    {
        private readonly ConcurrentQueue<MqttQueueItem> _messageQueue = new ConcurrentQueue<MqttQueueItem>();
        private readonly SemaphoreSlim _signal = new SemaphoreSlim(0);
        private readonly CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource();
        private readonly MqttTopicLogger _mqttTopicLogger = MqttTopicLogger.Instance;
        private readonly int _processingIntervalMs;
        private readonly Func<string, string, string, string, Task> _sendMqttFunc;
        private bool _isProcessing = false;

        /// <summary>
        /// 消息队列是否正在运行
        /// </summary>
        public bool IsRunning => _isProcessing;

        public MqttMessageQueue(int processingIntervalMs, Func<string, string, string, string, Task> sendMqttFunc)
        {
            _processingIntervalMs = processingIntervalMs;
            _sendMqttFunc = sendMqttFunc;
        }

        public void Enqueue(string source, string topic, string type, string messageJson)
        {
            _messageQueue.Enqueue(new MqttQueueItem
            {
                Source = source,
                Topic = topic,
                Type = type,
                MessageJson = messageJson
            });
            _signal.Release();

            if (!_isProcessing)
            {
                StartProcessing();
            }
        }

        private void StartProcessing()
        {
            if (_isProcessing)
                return;

            _isProcessing = true;
            Task.Run(async () => await ProcessQueueAsync(_cancellationTokenSource.Token));
        }

        private async Task ProcessQueueAsync(CancellationToken cancellationToken)
        {
            try
            {
                while (!cancellationToken.IsCancellationRequested)
                {
                    await _signal.WaitAsync(cancellationToken);

                    if (_messageQueue.TryDequeue(out var item))
                    {
                        try
                        {
                            await _sendMqttFunc(item.Source, item.Topic, item.Type, item.MessageJson);
                        }
                        catch (Exception ex)
                        {
                            // 记录发送失败的异常，但继续处理队列
                            _mqttTopicLogger.WriteTopicLog(item.Topic, $"从队列发送MQTT消息失败: {ex.Message}", "ERROR");
                        }

                        // 添加延迟以控制发送频率
                        await Task.Delay(_processingIntervalMs, cancellationToken);
                    }
                }
            }
            catch (OperationCanceledException)
            {
                // 正常取消操作，不需要处理
            }
            catch (Exception ex)
            {
                _mqttTopicLogger.WriteSystemError($"处理MQTT消息队列时出错: {ex.Message}", ex);
            }
            finally
            {
                _isProcessing = false;

                // 如果队列中还有消息，重新启动处理
                if (!_messageQueue.IsEmpty)
                {
                    StartProcessing();
                }
            }
        }

        public void Stop()
        {
            _cancellationTokenSource.Cancel();
        }
    }

    public class MqttQueueItem
    {
        public string Source { get; set; }
        public string Topic { get; set; }
        public string Type { get; set; }
        public string MessageJson { get; set; }
    }
}
